package com.toutiao.issue

import scala.collection.mutable.ArrayBuffer
import scala.collection.mutable.{HashMap => HMap}

import scala.math.sqrt

import java.util.Date
import java.net.URL

import org.apache.log4j.Logger

// 聚类上的聚类, 支持变化
// 但聚类Center不变化
// 虽然仍然不完美，但是应该是可以接受的

class IssueCollector(val TopS:Double = 0.25) {
  
  private
  val _log = Logger.getLogger(this.getClass.getName)
  
  private
  def getNumber(ms:Array[String]) = {
    def getHost(u:String) = {
      try {
        new URL(u).getHost()
      }
      catch {
        case _:Throwable => u
      }
    }
    val us = ms.map(getHost)
    us.toSet.size
  }
  
  
  private
  val K = 24
  private
  val Index = new Array[InvertIndex[String]](K)
  for (j <- 0 until K){
    Index(j) = new InvertIndex[String]
  }
  def insertIndex(i:Int, vs:Array[String]) {
    val j = i%K
    Index(j).insert(i,vs)
  }
  
  def removeIndex(i:Int, vs:Array[String]) {
    val j = i%K
    Index(j).remove(i,vs)
  }
  
  var data = Map[String,Event]()
  def size = data.size
  def apply(name:String) = data(name)
  def isIn(name:String) = data.contains(name)
  
  def keySet = data.keySet
  
  def insertData(name:String,event:Event) {
    data += name->event
  }
  
  private
  def cos(m1:Map[String,Double], s1:Double, m2:Map[String,Double], s2:Double):Double = {
    var s = 0.0
    for ( (k,v) <- m1){
      s += v*m2.getOrElse(k,0.0)
    }
    s/(s1*s2)
  }
  
  def remove(name:String,i:Int) {
    if (data.contains(name)){
    	val vs = data(name).vs.map(_._1).toArray
    	data -= name
    	removeIndex(i,vs)
    }
  }
  
  private
  def retrieve(vs:Map[String,Double],S:Double) = {
    val us = vs.filter(_._2 > 0.005)
    //_log.info(vs.size+" "+us.size)
    val es = (0 until K).map(j => {
      new WAND(Index(j),us,S)
    })
    // !!!!! 
    val ns = es.par.flatMap(e => e.retrieve)
    ns
  }

  def updateMs(name:String, x:Event) = {
	  if (data.contains(name)){
		  val e = data(name)
		  val ms = (e.ms ++ x.ms).toSet.toArray
		  val n = ms.size-e.ms.size
		  val vs = 
		    if (n > 0 && e.vs.size > 0 && x.vs.size > 0){
		      merge(e.vs,e.ms.size,x.vs,n)
		    }
		    else {
		      e.vs
		    }
         val t1 =
        		 if (e.T1.getTime < x.T1.getTime) e.T1
        		 else x.T1
         val t2 = 
                 if (e.T2.getTime < x.T2.getTime) x.T2 
                 else e.T2
		  data -= name
		  data += name->(new Event(name,ms,vs,t1,t2,e.s))    
	  }
  }

  private
  def cut(vs:Map[String,Double],size:Int=100) = {
    if (vs.size <= size){
      vs
    }
    else {
      val us = vs.toArray.sortBy(-_._2)
      val ss = us(size)._2
      val ns = vs.filter(_._2 > ss)
      val s = ns.map(x => x._2*x._2).reduce(_+_)
      ns.map(x => {(x._1,x._2/s)})
    }
  }
  private
  def merge(m1:Map[String,Double],n1:Int,m2:Map[String,Double],n2:Int) = {
    val n = n1+n2
    var z = m1
    for ( (k,v) <- m2){
      var x = v*n2+z.getOrElse(k,0.0)*n1
      z += k->x/n
    }
    var s = z.map(x => x._2*x._2).reduce(_+_)
    s = sqrt(s)
    z.map( x => {
      (x._1,x._2/s)
    })
    cut(z)
  }
  
  def processNewOne(
		  M1:HMap[String,Int],
		  M2:HMap[Int,String],
		  x:Event,
		  S:Double,
		  minCnt:Int) = {
    var event = x
    var r = _processOne(M1,M2,event,S,minCnt)
    var rs = List[String]()
    if (r == event.name){
      rs = rs :+ r
    }
    else {
      while ( r != ""){
        rs = rs :+ r
        event = data(r)
        r = _processOne(M1,M2,event,S,minCnt)
      }
    }
    
    rs
  }

  def _processOne(M1: HMap[String,Int],
		  		  M2: HMap[Int,String],
		  		  event: Event,
		  		  S: Double,
		  		  minCnt:Int) = {
    //_log.info(event.name)
    val ns = retrieve(event.vs,S).
    			filter(M2.contains(_)).
    			map(M2(_)).
                filter(data.contains(_)).
    			filter( _ != event.name)
    // !!!
    //_log.info(ns.size)
    val rs = ns.par.map(n => {
      (n, cos(data(n).vs, data(n).s, event.vs, event.s))
    })
    val (z, s) = if (rs.size > 0) rs.maxBy(_._2) else ("",0.0)
    //_log.info(s)
    if (s >= S){
      //_log.info(event.name+" "+z+" "+s)
      // Similar
      val eventD = data(z)
      val sizeD = eventD.ms.size // Merge
      val vs = merge(eventD.vs,sizeD, event.vs, event.ms.size) 
      val g = eventD.s
      val ms = eventD.ms ++ event.ms
      val t1 =
        if (eventD.T1.getTime < event.T1.getTime) eventD.T1
        else event.T1
      val t2 = 
        if (eventD.T2.getTime < event.T2.getTime) event.T2 
        else eventD.T2
 
      val index = M1(z)
      if (data.contains(z)){
        removeIndex(index,data(z).vs.map(_._1).toArray)
        data -= z
      }
      // 变成空集，也许直接删除更好？
      if (data.contains(event.name)){
        removeIndex(M1(event.name),event.vs.map(_._1).toArray)
        data -= event.name
        data += event.name->( 
            new Event(event.name, Array(z+" "+z+" "+z),Map[String,Double](), // 表面event.name合并到了z
                t1,t2,0.0))
      }
      data += z->(new Event(z,ms,vs,t1,t2,g))
      insertIndex(index,data(z).vs.map(_._1).toArray)
      
      z
    }
    else {  
      if (!data.contains(event.name)){
        // Become New Event
        val ms = event.ms.flatMap(u => u.split(" ")).toSet.toArray
        val msSize = ms.size
        if (msSize >= minCnt){
    	  data += event.name->event
    	  val i = M1(event.name) 
    	  insertIndex(i,event.vs.map(_._1).toArray)
    	  event.name  
        } 
        else {
          ""
        }
      }
      // Or Exist
      else {
        ""
      }
    }
    
  }
}
